package com.wdl.market.utils

import com.wdl.market.bean.MarketUserBehavior
import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}

import java.util.UUID
import scala.util.Random

/**
 * 自定义测试输入数据源
 */
class SimulateMarketEventSource extends RichParallelSourceFunction[MarketUserBehavior]{

  /** 定义是否在运行的标识位 */
  var running: Boolean = true

  /** 定义用户行为和渠道推广的集合 */
  val channelSet: Seq[String] = Seq("appStore", "huawei", "weibo", "wechat")
  val behaviorSet: Seq[String] = Seq("CLICK", "DOWNLOAD", "INSTALL", "UNINSTALL")

  /** 定义随机数生成器 */
  val rand: Random = Random

  override def run(ctx: SourceFunction.SourceContext[MarketUserBehavior]): Unit = {

    /** 定义一个发出数据的最大量，用于控制测试数据量 */
    val maxCounts: Long = Long.MaxValue
    var count = 0L

    /** while 循环，不停地随机生成数据 */
    while (running && (count < maxCounts)){
      val id: String = UUID.randomUUID().toString
      val behavior: String = behaviorSet(rand.nextInt(behaviorSet.size))
      val channel: String = channelSet(rand.nextInt(channelSet.size))
      val ts: Long = System.currentTimeMillis()

      ctx.collect(MarketUserBehavior(id, behavior, channel, ts))

      count += 1
      Thread.sleep(50L)

    }

  }

  override def cancel(): Unit = {

  }
}
